package cn.colins.raft.log;

import cn.colins.raft.LogManage;
import cn.colins.raft.enity.LogEntity;
import cn.colins.raft.enity.LogIndexPullResult;
import cn.colins.raft.enity.RaftNodeInfo;
import cn.colins.raft.enums.NodeStatusEnums;
import cn.colins.raft.exception.RaftRunException;
import cn.colins.raft.node.RaftNodeService;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @Description 非常简单而又丑陋的数据日志管理
 * @Author czl
 * @Version V1.0.0
 * @Since 1.0
 * @Date 2023/7/19
 */
public class SimpleLogManage implements LogManage {
    private static final Logger log = LoggerFactory.getLogger(SimpleLogManage.class);

    /**
     * 数据在内存中的副本
     */
    private Map<String, String> dataMap = new ConcurrentHashMap<>(1024);

    /**
     * 待提交的日志缓存
     */
    private ConcurrentHashMap<Long, LogEntity> cacheLogMap = new ConcurrentHashMap<>(1024);

    /**
     * 缓存日志id
     */
    private AtomicLong cacheLogId = new AtomicLong(-1L);

    /**
     * 检查点==行数
     */
    private volatile long checkPoint = 0L;

    private final String logPath;

    private RandomAccessFile file;

    public SimpleLogManage(String logPath) {
        this.logPath = logPath;
        try {
            this.file = new RandomAccessFile(logPath, "rw");
        } catch (FileNotFoundException e) {
            throw new RaftRunException(String.format("%s file not found", logPath));
        }
    }


    private long getNextCacheLogId() {
        return cacheLogId.incrementAndGet();
    }


    @Override
    public long preCommitLog(LogEntity logEntity) {
        // 1.放入待提交的日志 此时还没有日志索引
        long nextCacheLogId = getNextCacheLogId();
        preCommitLog(nextCacheLogId, logEntity);
        return nextCacheLogId;
    }

    @Override
    public void preCommitLog(long preCommitLogId, LogEntity logEntity) {
        cacheLogMap.put(preCommitLogId, logEntity);
    }

    @Override
    public void cacheLogRemove(long cacheLogId) {
        cacheLogMap.remove(cacheLogId);
    }

    @Override
    public synchronized long commitLog(long cacheLogId) {

        LogEntity logEntity = cacheLogMap.get(cacheLogId);
        try {
            long length = file.length();
            // 点位如果不是最后的话，需要放到最后 防止从follow切到leader后点位不是最后
            if (length != file.getFilePointer()) {
                file.seek(length);
            }
            // 一条日志控制在512字节内
            // 索引+1
            logEntity.setIndex((length / 512) + 1);
            logWriteFile(logEntity);
        } catch (Exception e) {
            log.error("[{}] : log commit fail: {}", logEntity.getCommand(), e.getMessage(), e);
            return 0L;
        } finally {
            // 清除缓存的日志
            cacheLogRemove(cacheLogId);
        }
        // 更新已提交索引
        RaftNodeInfo.getInstance().setCommitLastLogIndex();
        // 将数据添加到内存
        dataHandler(logEntity.getCommand());
        return logEntity.getIndex();
    }

    @Override
    public synchronized void commitLog(long cacheLogId, long logIndex) {
        // 上面的提交是针对leader的，leader一直顺序写所以不需要调整文件游标
        // 这里需要根据leader传来的索引，来确定游标的位置 index*512 -512 就是写入的点位
        LogEntity logEntity = cacheLogMap.get(cacheLogId);
        // 这里有可能取出为null的
        if (logEntity != null) {
            logEntity.setIndex(logIndex);
            try {
                file.seek(logIndex * 512 - 512);
                logWriteFile(logEntity);
            } catch (IOException e) {
                log.error("[{}] : log commit fail: {}", logEntity.getCommand(), e.getMessage(), e);
            } finally {
                // 清除缓存的日志
                cacheLogRemove(cacheLogId);
            }
            // 将数据添加到内存
            dataHandler(logEntity.getCommand());
        }
        // 不管是否成功刷盘，这里都需要记录从leader同步的索引
        RaftNodeInfo.getInstance().setLastLogIndex(logIndex);
    }

    @Override
    public void logIndexCheck() {
        long lastLogIndex = RaftNodeInfo.getInstance().getLastLogIndex();
        log.debug("日志check开始，checkPoint:{} lastLogIndex:{}", this.checkPoint, lastLogIndex);
        // 只有follow才需要检测
        if (checkPoint < lastLogIndex) {

            // 需要拉取的日志索引
            List<Long> pullLogIndex = new ArrayList<>();
            RandomAccessFile file = null;
            try {
                file = new RandomAccessFile(logPath, "rw");

                long startCheckPoint = checkPoint == 0L ? 1 : checkPoint;
                // 从checkPoint开始遍历到最后的提交的索引处
                for (; startCheckPoint <= lastLogIndex; startCheckPoint++) {
                    long filePoint = (startCheckPoint - 1) * 512;
                    // 不能超过文件上限
                    if (filePoint >= file.length()) {
                        log.debug("日志缺失行数索引：{}", startCheckPoint);
                        pullLogIndex.add(startCheckPoint);
                        continue;
                    }
                    file.seek(filePoint);
                    byte[] bytes = new byte[11];
                    file.read(bytes);
                    long logIndex = file.readLong();
                    if (logIndex == 0L) {
                        log.debug("日志缺失行数索引：{}", startCheckPoint);
                        pullLogIndex.add(startCheckPoint);
                    }
                }
                if (CollectionUtil.isEmpty(pullLogIndex)) {
                    // 为空说明没有需要拉的，更新check点位
                    this.checkPoint = lastLogIndex;
                    return;
                }
                LogIndexPullResult logIndexPullResult = RaftNodeService.sendLogPullRequest(pullLogIndex);
                log.debug("拉回的日志数据：{}", JSONObject.toJSON(logIndexPullResult));
                pullLogIndexHandler(logIndexPullResult);

            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                fileClose(file);
            }

        }
    }

    @Override
    public LogEntity getLogEntityByIndex(long logIndex, RandomAccessFile file) {
        try {
            file.seek(logIndex * 512);
            byte[] bytes = new byte[5];
            file.read(bytes);
            long term = file.readLong();
            byte[] bytes1 = new byte[6];
            file.read(bytes1);
            long index = file.readLong();
            if (index == 0L) {
                return null;
            }
            byte[] bytes2 = new byte[8];
            file.read(bytes2);
            byte[] bytes3 = new byte[476];
            file.read(bytes3);
            String command = new String(bytes3);
            return new LogEntity(term, index, command.trim());
        } catch (IOException e) {
            return null;
        }
    }

    /**
     * 从leader拉过来的数据处理
     */
    private void pullLogIndexHandler(LogIndexPullResult logIndexPullResult) {
        if (null != logIndexPullResult) {
            for (LogEntity logEntity : logIndexPullResult.getLogEntities()) {
                try {
                    file.seek(logEntity.getIndex() * 512 - 512);
                    logWriteFile(logEntity);
                    // 数据添加到内存
                    dataHandler(logEntity.getCommand());
                } catch (IOException e) {
                    log.error("[{}] : log commit fail: {}", logEntity.getCommand(), e.getMessage(), e);
                } finally {
                    // 每提交一个，点位下移
                    checkPoint++;
                }
            }
        }
    }

    /**
     * 日志数据写入文件
     */
    private void logWriteFile(LogEntity logEntity) throws IOException {
        file.write("term:".getBytes());       //5字节
        file.writeLong(logEntity.getTerm());  //8字节
        file.write("index:".getBytes());      //6字节
        file.writeLong(logEntity.getIndex()); //8字节
        file.write("command:".getBytes());    //8字节
        ByteBuffer allocate = ByteBuffer.allocate(476);
        allocate.put(logEntity.getCommand().getBytes());
        file.write(allocate.array());         //476字节
        file.write("\n".getBytes());          //1字节
    }

    @Override
    public void init() {
        try {
            dataInit();
            if (file.length() > 0) {
                long lastLog = file.length() - 512;
                file.seek(lastLog);
                byte[] termByte = new byte[5];
                file.read(termByte);
                long term = file.readLong();
                RaftNodeInfo.getInstance().setCurrentTerm(term);
                byte[] indexByte = new byte[6];
                file.read(indexByte);
                long logIndex = file.readLong();
                RaftNodeInfo.getInstance().setLastLogIndex(logIndex);
                RaftNodeInfo.getInstance().setLastLogTerm(term);
                file.seek(file.length());
                // 设置缓存的日志id也从这里索引处开始自增
                this.cacheLogId = new AtomicLong(logIndex);
                // 数据初始化完了，check点还为0 说明没有日志缺失断层，所以直接等于最后的日志索引即可
                this.checkPoint = this.checkPoint == 0L ? logIndex : this.checkPoint;
            }
            RaftNodeInfo.getInstance().setLogManage(this);
        } catch (IOException e) {
            log.error("Log 初始化失败:{}", e.getMessage(), e);
            throw new RaftRunException("Log 初始化失败");
        }
    }

    private void dataInit() throws IOException {
        log.debug("----------------------------------日志文件数据加载中-----------------------------------");
        for (long i = 0, index = file.length() / 512; i < index; i++) {
            LogEntity logEntity = this.getLogEntityByIndex(i, this.file);
            if (logEntity == null && this.checkPoint == 0L) {
                // 第一次找到空缺的位置需要记录下来，之后check直接从这里开始
                this.checkPoint = i + 1;
            }
            if (logEntity != null) {
                log.debug("加载日志文件数据：{}", JSONObject.toJSON(logEntity));
                dataHandler(logEntity.getCommand());
            }
        }
        log.debug("----------------------------------日志文件数据加载完成-----------------------------------");
    }

    public void dataHandler(String command) {
        //解析语法并缓存到内存
        String[] data = command.split(" ");
        if (data[0].equals("SET") && data.length == 3) {
            dataMap.put(data[1], data[2]);
        }
    }

    @Override
    public void destroy() {
        fileClose(this.file);
    }

    private void fileClose(RandomAccessFile file) {
        if (file != null) {
            try {
                file.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public String getDataByKey(String key) {
        return dataMap.get(key);
    }

}
